package com.cn.namesrv;

import com.cn.coder.MessageDecoder;
import com.cn.coder.MessageEncoder;
import com.cn.common.processor.INettyRequestProcessor;
import com.cn.common.Pair;
import com.cn.msg.Message;
import com.cn.remote.AbStractNetty;
import com.cn.remote.ChannelListener;
import com.cn.remote.INettyRemoteService;
import com.cn.remote.RemoteUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultEventExecutorGroup;

import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class NameSrvServer extends AbStractNetty implements INettyRemoteService {

    /**
     * 工作线程数
     */
    private int worker_pool_size = 8;

    /**
     * 服务端口
     */
    private int port = 1010;

    private String host = "0.0.0.0";


    private ServerBootstrap bootstrap = null;

    private EventLoopGroup boss = null;

    private EventLoopGroup worker = null;

    private ChannelListener channelListener;


    public NameSrvServer(int worker_pool_size, int port, String host, ChannelListener channelListener) {
        super(500);
        this.channelListener = channelListener;
        this.callbackExecutor = Executors.newFixedThreadPool(16);

        this.worker_pool_size = worker_pool_size > 0 ? worker_pool_size : this.worker_pool_size;
        this.port = port;
        this.host = host;

        this.bootstrap = new ServerBootstrap();
        boolean linux = Epoll.isAvailable();

        if (linux) {
            this.boss = new EpollEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("boss线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        } else {
            this.boss = new NioEventLoopGroup(1, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("boss线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        }

        //工作线程
        if (linux) {
            this.worker = new EpollEventLoopGroup(this.worker_pool_size, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("worker线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        } else {
            this.worker = new NioEventLoopGroup(this.worker_pool_size, new ThreadFactory() {
                private AtomicInteger atomicInteger = new AtomicInteger(0);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(r, String.format("worker线程：%d", this.atomicInteger.incrementAndGet()));
                }
            });
        }


        DefaultEventExecutorGroup defaultEventExecutorGroup = new DefaultEventExecutorGroup(this.worker_pool_size, new ThreadFactory() {
            private AtomicInteger atomicInteger = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, String.format("ServerCodecThread_%d", this.atomicInteger.incrementAndGet()));
            }
        });

        linux = false;
        this.bootstrap.group(boss, worker);
        //linux用 EpollServerSocketChannel ；window用NioServerSocketChannel
        if (!Epoll.isAvailable()) {
            bootstrap.channel(NioServerSocketChannel.class);
        } else {
            bootstrap.channel(EpollServerSocketChannel.class);
        }

        this.bootstrap.option(ChannelOption.SO_BACKLOG, 1024)
                .option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_KEEPALIVE, false)
                .childOption(ChannelOption.TCP_NODELAY, true)
                .childOption(ChannelOption.SO_SNDBUF, 65535)// 发送缓冲大小
                .childOption(ChannelOption.SO_RCVBUF, 65535);// 接收缓冲大小
        this.bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                //如果是自定义消息，那么用以下代码
                ch.pipeline().addLast(defaultEventExecutorGroup,
//                        new LoggingHandler(LogLevel.ERROR),//日志
                        new MessageEncoder(),//编码器
                        new MessageDecoder(),//解码器
                        new IdleStateHandler(0, 0, 6, TimeUnit.SECONDS),//心跳检测，不管读写，只要超过6秒钟，就主动释放客户端占用的资源
                        new NameSrvConnectManageHandler(),
                        new NameSrvServiceHandler()//业务处理
                );

            }
        });
    }

    public void start() {
        if (bootstrap == null) throw new RuntimeException("初始化失败！");
        try {
            //绑定端口，开始同步
            ChannelFuture channel = bootstrap.bind(host, port).sync();
            InetSocketAddress addr = (InetSocketAddress) channel.channel().localAddress();
            this.port = addr.getPort();
            System.out.println("nameSrv节点启动成，地址：" + addr.getAddress().getHostAddress() + "，端口：" + addr.getPort());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void stop() {
        if (this.boss != null) {
            this.boss.shutdownGracefully();
            this.worker.shutdownGracefully();
            System.out.println("nameSrv服务已关闭！");
        }
    }

    /**
     * 注册
     *
     * @param code
     * @param pair
     */
    public void registerProcessor(int code, Pair<INettyRequestProcessor, ExecutorService> pair) {
        this.processTables.put(code, pair);
    }

    @Override
    public void unRegisterConnect(ChannelHandlerContext ctx) {
        this.channelListener.unregisterEvent(ctx);
    }

    @Override
    public void registerConnect(ChannelHandlerContext ctx) {
        this.channelListener.registerEvent(ctx);
    }


    class NameSrvServiceHandler extends SimpleChannelInboundHandler<Message> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
            process(msg, ctx);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            unRegisterConnect(ctx);
//            NameRemoveBroker.close(ctx);
            RemoteUtil.close(ctx);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            unRegisterConnect(ctx);
            cause.printStackTrace();
//            NameRemoveBroker.close(ctx);
            RemoteUtil.close(ctx);
        }
    }

}
